package com.sub.kafka.producer.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @ClassName SubProducerValueInterceptor
 * @Description: 自定义生产者拦截器
 * 1、实现ProducerInterceptor接口
 * 2、重写抽象方法
 * 主要：onSend方法，该方法用于实现拦截器逻辑，该方法会在消息发送前被调用，返回值是一个ProducerRecord对象，如果返回null，则消息不会被发送
 * @Author Submerge.
 * @Since 2025/6/1 20:37
 * @Version 1.0
 */
public class SubProducerValueInterceptor implements ProducerInterceptor<String, String> {

    private static final String SubValueInterceptor = "SubValueInterceptor:";

    /**
     * 数据发送前，会执行此方法，进行数据发送前的预处理
     * @param record the record from client or the record returned by the previous interceptor in the chain of interceptors.
     * @return
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(), SubValueInterceptor+ record.value());
    }

    /**
     * 数据发送后，获取应答时，会执行此方法
     * @param metadata
     * @param exception
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("发送成功");

    }

    /**
     * 生产者关闭时，会执行此方法，完成一些资源回收和释放的操作
     */
    @Override
    public void close() {
        System.out.println("关闭拦截器");

    }

    /**
     * 创建生产者对象的时候，会执行此方法，可以根据场景对生产者对象的配置进行统一修改或转换。
     * @param configs
     */
    @Override
    public void configure(Map<String, ?> configs) {
        System.out.println("配置拦截器");
        for (String s : configs.keySet()) {
            System.out.println(s + ":" + configs.get(s));
        }

    }
}
